package com.java.springcloud.listener;

import com.mongodb.client.model.changestream.ChangeStreamDocument;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.messaging.Message;
import org.springframework.data.mongodb.core.messaging.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * @author dushougudu
 */
@Component
@Slf4j
public class ChangeStreamDocumentListener implements MessageListener<ChangeStreamDocument<Document>, Document> {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(Message<ChangeStreamDocument<Document>, Document> message) {
        log.info("{} Received Message in collection: {},message raw: {}, message body:{}",
                Thread.currentThread().getId(),
                message.getProperties().getCollectionName(), message.getRaw(), message.getBody());
        String resumeToken = message.getRaw().getResumeToken().toJson();
        redisTemplate.opsForValue().set("resumeToken", resumeToken);
        String key = message.getRaw().getDocumentKey().getObjectId("_id").toString();
        kafkaTemplate.send("mongodb", key, message.getBody().toJson());
    }
}
